package com.example.springboot159.util;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.*;

public class ThreadPoolManager  {
    private static final Logger logger= LoggerFactory.getLogger(ThreadPoolManager.class);

    private static ThreadPoolManager tpm = new ThreadPoolManager();
    private static final int CORE_POOL_SIZE = 600;
    private static final int MAX_POOL_SIZE =  1000;
    private static final int KEEP_ALIVE_TIME = 0;
    private static final int WORK_QUEUE_SIZE = 300;
    private static final int TASK_QOS_PERIOD = 300;
    private Queue<Runnable> taskQueue = new LinkedList();

    //拒绝策略
    final RejectedExecutionHandler handler = new RejectedExecutionHandler() {
        public void rejectedExecution(Runnable task, ThreadPoolExecutor executor) {
            ThreadPoolManager.this.taskQueue.offer(task);
        }
   };

    final ThreadPoolExecutor threadPool = new ThreadPoolExecutor(CORE_POOL_SIZE,  MAX_POOL_SIZE, KEEP_ALIVE_TIME, TimeUnit.SECONDS,
            new ArrayBlockingQueue(WORK_QUEUE_SIZE), this.handler);


    //起一个定时任务，处理被拒绝的任务
    final Runnable accessBufferThread = new Runnable() {
        public void run() {
            if (ThreadPoolManager.this.hasMoreAcquire())
                ThreadPoolManager.this.threadPool.execute(ThreadPoolManager.this.taskQueue.poll());
        }
    };

    final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);

    final ScheduledFuture<?> taskHandler = this.scheduler.scheduleAtFixedRate(this.accessBufferThread, 0L, 300L, TimeUnit.MILLISECONDS);
    public static ThreadPoolManager newInstance()
    {
        return tpm;
    }

    public static void addThread(Runnable task) {
        tpm.addExecuteTask(task);
    }

    private boolean hasMoreAcquire()
    {
        return !this.taskQueue.isEmpty();
    }

    public void addExecuteTask(Runnable task)
    {
        if (task != null)
            this.threadPool.execute(task);
    }

    public static void shutdown()
    {
        tpm.threadPool.shutdown();
        try
        {
            //awaitTermination检测线程池在timeout时间后，是否关闭，如果为关闭返回true
            if (!tpm.threadPool.awaitTermination(60L, TimeUnit.SECONDS)) {
                tpm.threadPool.shutdownNow();
                logger.warn("Interrupt the worker, which may cause some task inconsistent. Please check the biz logs.");

                if (!tpm.threadPool.awaitTermination(60L, TimeUnit.SECONDS))
                    logger.error("Thread pool can't be shutdown even with interrupting worker threads, which may cause some task inconsistent. Please check the biz logs.");
            }
        } catch (InterruptedException ie) {
            tpm.threadPool.shutdownNow();
            logger.error("The current server thread is interrupted when it is trying to stop the worker threads. This may leave an inconcistent state. Please check the biz logs.",ie);
            Thread.currentThread().interrupt();
        }
    }

}

